//package com.intct.hbase.func;
//
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.api.java.tuple.Tuple3;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.functions.async.ResultFuture;
//import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
//import org.apache.flink.util.concurrent.ExecutorThreadFactory;
//import org.apache.hadoop.hbase.TableName;
//import org.apache.hadoop.hbase.TableNotFoundException;
//import org.apache.hadoop.hbase.client.*;
//import org.apache.hadoop.hbase.util.Bytes;
//import org.apache.hadoop.hbase.util.Threads;
//
//import java.util.ArrayList;
//import java.util.List;
//import java.util.concurrent.CompletableFuture;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.function.BiConsumer;
//
///**
// * @author gufg
// * @since 2025-08-12 18:18
// */
//public class HBaseAyncJoinFun extends RichAsyncFunction<Tuple2<String,String>, Tuple3<String,String,String>> {
//    @Override
//    public void asyncInvoke(Tuple2<String, String> input, ResultFuture<Tuple3<String, String, String>> resultFuture) throws Exception {
//
//        final ExecutorService threadPool = Executors.newFixedThreadPool(2,
//                new ExecutorThreadFactory("hbase-aysnc-lookup-worker", Threads.LOGGING_EXCEPTION_HANDLER));
//
//        AsyncConnection connection = ConnectionFactory.createAsyncConnection().get();
//
//        AsyncTable<ScanResultConsumer> table = connection.getTable(TableName.valueOf(""), threadPool);
//
//        Get get =new Get(Bytes.toBytes(""));
//        CompletableFuture<Result> resultCompletableFuture = table.get(get);
//
//        resultCompletableFuture.whenCompleteAsync(new BiConsumer<Result, Throwable>() {
//            @Override
//            public void accept(Result result, Throwable throwable) {
//                if (throwable != null){
//                    if (throwable instanceof TableNotFoundException){
//                        resultFuture.completeExceptionally(
//                                new RuntimeException("HBase table '"+"table_name"+"' not found.",throwable)
//                        );
//                    }else {
//                        resultCompletableFuture.completeExceptionally(throwable);
//                    }
//                }else{
//                    List list = new ArrayList<Tuple3<String, String, String>>();
//                    if (result.isEmpty()){
//                        String cityName="";
//                        list.add(new Tuple3<String,String,String>(row_key,p_name,cityName));
//                        resultFuture.complete(list);
//                    }else{
//                        String cityName = Bytes.toString(result.getValue(Bytes.toBytes("cf"),Bytes.toBytes("city_name")));
//                        list.add(new Tuple3<String,String,String>(row_key,p_name,cityName));
//                        resultFuture.complete(list);
//                    }
//                }
//            }
//        });
//    }
//}
